from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum, regexp_extract, when, split, explode, array, coalesce, count, countDistinct
import csv, os
import pandas as pd
import matplotlib.pyplot as plt

spark = SparkSession.builder \
    .appName('data_processing') \
    .config('spark.default.parallelism', 300) \
    .getOrCreate()  # 创建sparkSession对象，builder是构建器，config的设置是并行度
plt.rcParams['font.sans-serif'] = ['SimHei']  # 指定使用中文字体，如微软雅黑、SimHei等


# 设置Spark配置变量spark.sql.crossJoin.enabled为true，或者使用CROSS JOIN语法来允许笛卡尔积操作
# .config("spark.sql.crossJoin.enabled", "true") \

# 分割时间
def GetTime(data):  # data为博物馆数据，包含时间即可
    time = data.select("预约到馆时间").collect()  # 单独取出预约到馆时间这一列
    time = [row["预约到馆时间"] for row in time]  # 转换为数组
    time = pd.DataFrame(time, columns=['datetime'])  # 转换为dataframe格式，并设置列名为datetime
    time = spark.createDataFrame(time)  # 转化安慰spark dataframe格式，这两个不一样
    # 用正则表达式除去前面的年月日，regexp_extract() 会从dataframe取出符合后面正则表达式的数据，
    # 然后创建一个time列来存储结果（withColumn），正则表达式中括号里的部分是会被提取的，1是索引值，就要提取哪部分，这里只有一个
    time = time.withColumn('time', regexp_extract('datetime', r'\d{4}-\d{2}-\d{2}(\d{2}:\d{2}-\d{2}:\d{2})', 1))
    time = time.select("time")  # 仅返回time列
    return time


# 返回上下、午预约数
def MANum(time):  # time为分割好的时间
    # 将时间段拆分成起始时间和结束时间
    time = time.withColumn('start_time', split(time['time'], '-').getItem(0))
    time = time.withColumn('end_time', split(time['time'], '-').getItem(1))
    # 添加上午和下午的标识列，when这边的意思是条件真返回1，其它0
    time = time.withColumn('morning', when(time['start_time'] <= '12:00', 1).otherwise(0))
    time = time.withColumn('afternoon', when(time['end_time'] >= '12:00', 1).otherwise(0))
    # 统计上午和下午的数量，alias用于命名，agg用于聚合数据
    session_count = time.agg(
        sum(when(time['morning'] == 1, 1).otherwise(0)).alias('morning_count'),
        sum(when(time['afternoon'] == 1, 1).otherwise(0)).alias('afternoon_count')
    )
    # 取出结果
    counts = session_count.first()
    # 取出上午和下午的数量
    morning = counts['morning_count']
    afternoon = counts['afternoon_count']
    return morning, afternoon


# 返回不同时间段预约数
def MANumMore(time):
    # 将时间段拆分成起始时间和结束时间
    time = time.withColumn('start_time', split(time['time'], '-').getItem(0))
    time = time.withColumn('end_time', split(time['time'], '-').getItem(1))
    # 定义时间段列表
    time_periods = ['6:00-10:00', '10:00-13:00', '13:00-16:00', '16:00-21:00']
    # 初始化时间段计数字典
    counts = {period: 0 for period in time_periods}
    # 遍历每个时间段
    for period in time_periods:
        # 提取时间段的起始时间和结束时间
        start_time, end_time = period.split('-')
        # 计算当前时间段的预约数量，filter用于筛选满足条件的行
        period_count = time.filter((time['start_time'] >= start_time) & (time['start_time'] <= end_time)).count()
        # 将当前时间段的预约数量累加到计数字典中
        counts[period] += period_count
    # 将时间段计数转为列表
    counts = list(counts.values())
    return counts


# 检查文件是否存在，不用每次都重新写一次
modified_csv_file = 'GY预约_modified.csv'
if not os.path.exists(modified_csv_file):
    # 随同人员这一列的数据自带“,”会导致pyspark读取csv异常，所以把这里的“,”替换为“:”
    with open('GY预约.csv', 'r', encoding='gbk') as file:
        csv_reader = csv.reader(file)
        headers = next(csv_reader)  # 获取列名
        rows = list(csv_reader)  # 获取数据行

    # 预处理随同人员数据
    for row in rows:
        row_index = headers.index('随同人员')
        companion_data = row[row_index]
        if companion_data != 'null':
            row[row_index] = companion_data.replace(',', ';')  # 将逗号替换为分号

    # 保存修改后的数据到新的CSV文件
    with open('GY预约_modified.csv', 'w', newline='', encoding='gbk') as file:
        csv_writer = csv.writer(file)
        csv_writer.writerow(headers)
        csv_writer.writerows(rows)

data = spark.read.csv('GY预约_modified.csv', inferSchema=True, header=True, encoding='gbk')

# 根据博物馆ID将数据分开,用于后面分析不同博物馆的人员情况
id = data.select("博物馆ID").distinct().collect()
id = [row["博物馆ID"] for row in id]  # 因为id是row，所以要取值后才能用
# print(len(id))
data1 = data.filter(col("博物馆ID") == id[0])
data2 = data.filter(col("博物馆ID") == id[1])

# 统计两个博物馆的总参观人数
tol1 = data1.select(sum("参观人数")).collect()[0][0]
tol2 = data2.select(sum("参观人数")).collect()[0][0]
tol1 = int(tol1)
tol2 = int(tol2)
# 显示结果
bookings = [tol1, tol2]
museums = [data1.select("博物馆名称").first()[0], data2.select("博物馆名称").first()[0]]
plt.bar(museums, bookings)
# 添加标题和标签
plt.title('博物馆预约人数')
plt.xlabel('博物馆名称')
plt.ylabel('预约人数')
plt.show()

# 博物馆参观时间段分布：上午，下午的预约数分布情况，这部分跑的比较慢，写其它的时候建议注释掉
time = GetTime(data)
# 统计上下午人数
morning, afternoon = MANum(time)
# 显示结果
booking_counts = [morning, afternoon]
time_periods = ['上午', '下午']
# 创建柱状图
plt.bar(time_periods, booking_counts)
# 添加标题和标签
plt.title('上午和下午预约人数分布')
plt.xlabel('时间段')
plt.ylabel('预约人数')
plt.show()

# 博物馆预约的人的年龄分段（少年0-17）（青年18-40）（中年41-65）（老年65以上）
# 定义年龄分段的顺序
age_order = ["少年", "青年", "中年", "老年"]

# 博物馆预约的人的年龄分段（少年0-17）（青年18-40）（中年41-65）（老年65以上）
# 创建新的年龄分段列
data = data.withColumn("年龄分段", when(data["年龄"].between(0, 17), "少年")
                       .when(data["年龄"].between(18, 40), "青年")
                       .when(data["年龄"].between(41, 65), "中年")
                       .when(data["年龄"] > 65, "老年")
                       .otherwise("未知"))

# 年龄分段分布统计
age_segment_counts = data.groupBy("年龄分段").count().orderBy("年龄分段")
age_segment_counts.show()

# 调整年龄分段的顺序
age_segment_counts = age_segment_counts.withColumn("age_order", when(age_segment_counts["年龄分段"] == "少年", 1)
                                                   .when(age_segment_counts["年龄分段"] == "青年", 2)
                                                   .when(age_segment_counts["年龄分段"] == "中年", 3)
                                                   .when(age_segment_counts["年龄分段"] == "老年", 4))
age_segment_counts = age_segment_counts.orderBy("age_order")

# 将结果转换为Pandas DataFrame并绘制图表
age_segment_counts_pd = age_segment_counts.toPandas()
age_segment_counts_pd = age_segment_counts_pd.set_index("年龄分段")
age_segment_counts_pd = age_segment_counts_pd.reindex(age_order)

# 绘制图表
ax = age_segment_counts_pd.plot(kind="bar")
ax.legend(["人数"])
plt.title("博物馆预约年龄分段分布")
plt.xlabel("年龄分段")
plt.ylabel("人数")
plt.show()

# 博物馆预约的人的性别分布
gender_counts = data.groupBy("性别").count()

# 将结果转换为Pandas DataFrame
gender_counts_pd = gender_counts.toPandas()

# 绘制性别分布图表
# 博物馆预约的人的性别分布统计
gender_counts = data.groupBy("性别").count()

# 将结果转换为Pandas DataFrame
gender_counts_pd = gender_counts.toPandas()

# 绘制性别分布饼图
plt.pie(gender_counts_pd["count"], labels=gender_counts_pd["性别"], autopct='%1.1f%%')
plt.title("博物馆预约性别分布")
plt.legend(gender_counts_pd["性别"])
plt.show()

# 不同博物馆相同时间参观人数分布,跑一遍要好久好久
time1 = GetTime(data1)
time2 = GetTime(data2)
time_periods = ['6:00-10:00', '10:00-13:00', '13:00-16:00', '16:00-21:00']
counts1 = MANumMore(time1)
counts2 = MANumMore(time2)
print(counts1)
print(counts2)
# 绘制折线图
plt.plot(time_periods, counts1, label=data1.select("博物馆名称").first()[0])
plt.plot(time_periods, counts2, label=data2.select("博物馆名称").first()[0])
# 添加标题和标签
plt.title('博物馆不同时间段预约人数分布')
plt.xlabel('时间段')
plt.ylabel('预约人数')
# 添加图例
plt.legend()
# 在每个折线的端点上标注数字
for i in range(len(time_periods)):
    plt.text(time_periods[i], counts1[i], str(counts1[i]), ha='center', va='top')
    plt.text(time_periods[i], counts2[i], str(counts2[i]), ha='center', va='bottom')
# 显示图表
plt.show()

# 预约时间和取消率的关系

# 获取预约时间数据
time_data = GetTime(data)

# 将time_data转换为Pandas DataFrame
time_data_pd = time_data.toPandas()

# 拆分预约到馆时间段为开始时间和结束时间
time_data_pd["开始时间"] = time_data_pd["time"].str.split("-").str[0]
time_data_pd["结束时间"] = time_data_pd["time"].str.split("-").str[1]

# 将拆分后的数据转换为Spark DataFrame
time_data = spark.createDataFrame(time_data_pd)

# 将拆分后的数据与原数据进行关联
data = data.join(time_data)

# 根据核销状态进行分类并计算取消率
cancel_rates = data.groupBy("开始时间", "结束时间").agg(
    (sum(when(data["核销状态"] == "已取消", 1)) / count("*")).alias("取消率"))

# 将结果转换为Pandas DataFrame
cancel_rates_pd = cancel_rates.toPandas()

# 构造时间段列
cancel_rates_pd["时间段"] = cancel_rates_pd["开始时间"] + "-" + cancel_rates_pd["结束时间"]

# 绘制预约时间段和取消率关系图
cancel_rates_pd.plot(kind="bar", x="时间段", y="取消率")
plt.title("预约时间段和取消率关系")
plt.xlabel("时间段")
plt.ylabel("取消率")
plt.legend(["取消率"])
plt.show()

# 拆分预约到馆时间段为开始时间和结束时间
data = data.withColumn("开始时间", split(data["预约到馆时间"], "-")[0])
data = data.withColumn("结束时间", split(data["预约到馆时间"], "-")[1])

# 根据核销状态进行分类并计算取消率
cancel_rates = data.groupBy("开始时间", "结束时间").agg(
    (sum(when(data["核销状态"] == "已取消", 1)) / count("*")).alias("取消率"))

# 将结果转换为Pandas DataFrame
cancel_rates_pd = cancel_rates.toPandas()

# 构造时间段列
cancel_rates_pd["时间段"] = cancel_rates_pd["开始时间"] + "-" + cancel_rates_pd["结束时间"]

# 绘制预约时间段和取消率关系图
cancel_rates_pd.plot(kind="bar", x="时间段", y="取消率")
plt.title("预约时间段和取消率关系")
plt.xlabel("时间段")
plt.ylabel("取消率")
plt.legend(["取消率"])
plt.show()
